/*
 * Copyright 2013-2015 the original author or authors.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *      http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package com.ls.fw.config.zookeeper;

import java.util.Arrays;
import java.util.List;
import java.util.Properties;

import org.apache.commons.lang3.StringUtils;
import org.apache.curator.RetryPolicy;
import org.apache.curator.ensemble.EnsembleProvider;
import org.apache.curator.ensemble.exhibitor.DefaultExhibitorRestClient;
import org.apache.curator.ensemble.exhibitor.ExhibitorEnsembleProvider;
import org.apache.curator.ensemble.exhibitor.ExhibitorRestClient;
import org.apache.curator.ensemble.exhibitor.Exhibitors;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.apache.curator.retry.RetryNTimes;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;

/**
 * {@link org.springframework.boot.autoconfigure.EnableAutoConfiguration Auto-configuration}
 * that sets up Zookeeper discovery.
 *
 * @author Spencer Gibb
 * @since 1.0.0
 */
public class ZookeeperClientFactory {

	private static final Logger log = LoggerFactory.getLogger(ZookeeperClientFactory.class);

	@Autowired(required = false)
	private EnsembleProvider ensembleProvider;

	public EnsembleProvider getEnsembleProvider() {
		return ensembleProvider;
	}

	public void setEnsembleProvider(EnsembleProvider ensembleProvider) {
		this.ensembleProvider = ensembleProvider;
	}
	public CuratorFramework newInstance(ZookeeperProperties properties) throws Exception {
		return newInstance(properties, null);
	}
	public CuratorFramework newInstance(ZookeeperProperties properties, String namespace) throws Exception {
		CuratorFrameworkFactory.Builder builder = CuratorFrameworkFactory.builder()
				 ;
		if(StringUtils.isNotBlank(namespace)){
			builder.namespace(namespace);
		}
		if (this.ensembleProvider != null) {
			builder.ensembleProvider(this.ensembleProvider);
		} else {
			builder.connectString(properties.getConnectString());
		}
		RetryPolicy retryPolicy = new ExponentialBackoffRetry(properties.getBaseSleepTimeMs(),
				properties.getMaxRetries(),
				properties.getMaxSleepMs());
		CuratorFramework curator = builder.retryPolicy(retryPolicy).build();
		curator.start();
		log.info("blocking until connected to zookeeper for " + properties.getBlockUntilConnectedWait() + properties.getBlockUntilConnectedUnit());
		curator.blockUntilConnected(properties.getBlockUntilConnectedWait(), properties.getBlockUntilConnectedUnit());
		log.info("connected to zookeeper");
		return curator;
	}
	
    /** 
     * build provider,all of params from config-file 
     * @param properties 
     * @return 
     */  
    protected EnsembleProvider buildProvider(Properties properties) {  
        String servers = properties.getProperty("zk.host.rest.servers");   //hosts.servers = 127.0.0.1,127.0.0.2  
        if (servers == null || servers.isEmpty()) {  
            throw new IllegalArgumentException("zk.host.servers cant be empty");  
        }  
        List<String> hostnames = Arrays.asList(servers.split(","));  
        String port = properties.getProperty("zk.host.rest.port");  
        Integer restPort = 80;   //default  
        if (port != null) {  
            restPort = Integer.valueOf(port);  
        }  
        final String backupAddress = properties.getProperty("zk.host.backup");//127.0.0.1:2181  
        //if network is error,you should sepcify a backup zk-connectString  
        Exhibitors exhibitors = new Exhibitors(hostnames, restPort, new Exhibitors.BackupConnectionStringProvider() {  
            @Override  
            public String getBackupConnectionString() throws Exception {  
                return backupAddress;  
            }  
        });  
        //rest,as meaning of getting fresh zk-connectString list.  
        ExhibitorRestClient restClient = new DefaultExhibitorRestClient();  
        String restUriPath = properties.getProperty("zk.host.rest.path");  
        String period = properties.getProperty("zk.host.rest.period");  
        Integer pollingMs = 180000; //3 min  
        if (period != null) {  
            pollingMs = Integer.valueOf(period);  
        }  
        return new ExhibitorEnsembleProvider(exhibitors, restClient, restUriPath, pollingMs, new RetryNTimes(10, 1000));  
    }  
}
